package com.atguigu.api3

import java.util.concurrent.TimeUnit

import com.atguigu.api.SensorReading
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state._
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.{CheckpointingMode, TimerService}
import org.apache.flink.util.Collector

/**
 *
 * @description: 温度连续10s上升测试
 * @author: baojinlong 
 **/
object F05StateBarrier {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 设置一致性检查点配置
    environment.enableCheckpointing(1000)
    val checkpointConfig: CheckpointConfig = environment.getCheckpointConfig
    checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
    checkpointConfig.setCheckpointTimeout(6000)
    checkpointConfig.setMaxConcurrentCheckpoints(2)
    // 在两次ck之间触发的最小时间
    checkpointConfig.setMinPauseBetweenCheckpoints(500)
    // 默认用ck启用故障恢复(而不是savePoint)
    checkpointConfig.setPreferCheckpointForRecovery(true)
    // 允许checkpoint失败次数
    checkpointConfig.setTolerableCheckpointFailureNumber(3)
    // 重启策略
    environment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000))
    environment.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(5, TimeUnit.MINUTES), Time.of(5, TimeUnit.MINUTES)))

    val inputStream: DataStream[String] = environment.socketTextStream("localhost", 7777)
    val dataStream: DataStream[SensorReading] = inputStream.map {
      data => {
        val arr: Array[String] = data.split(",")
        SensorReading(arr(0), timestamp = arr(1).toLong, temperature = arr(2).toDouble)
      }
    }
    // 说明dataStream流中每次进入的数据为SensorReading
    val resultInfo: DataStream[String] = dataStream
      .keyBy(_.id)
      .process(new MyKeyedProcessFunctionDemo001(1000))
    resultInfo.print("resultInfo")
    environment.execute("state test")
  }

}

/**
 * 具体案例:连续上升数据则报警,输出为警告信息 key为字符串传感器idString
 */
class MyKeyedProcessFunctionDemo001(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
  // 保存上一个问题进行比较
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTempState", classOf[Double]))
  // 保存定时器id
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTsState", classOf[Long]))


  override def processElement(value: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    // 取出上一个温度状态
    val oldTemp: Double = lastTempState.value
    // 获取当前温度
    val nowTemperature: Double = value.temperature
    // 赋值
    lastTempState.update(nowTemperature)

    // 当前温度与上次温度比较为上升同时当时没有定时器
    // 获取上一个定时器数据
    val oldTimerTsState: Long = timerTsState.value
    // 获取定时器对象
    val timerService: TimerService = context.timerService
    if (nowTemperature > oldTemp && oldTimerTsState == 0) {
      // 注册当前时间10s的定时器
      val nowTimerTsState: Long = timerService.currentProcessingTime + interval
      // 更新定时器参数
      timerTsState.update(nowTimerTsState)
      // 注册定时器
      timerService.registerProcessingTimeTimer(nowTimerTsState)
    } else if (nowTemperature < oldTemp) {
      // 如果温度下降则删除定时器
      timerService.deleteProcessingTimeTimer(oldTimerTsState)
      // 清空状态或者置为0
      //timerTsState.update(0)
      timerTsState.clear()
    }

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    // 处理定时器逻辑
    out.collect("传感器" + ctx.getCurrentKey + "的温度连续上升" + interval / 1000 + "秒")
    // 清空定时器状态
    timerTsState.clear()
  }
}

class MyKeyedProcessFunction002 extends KeyedProcessFunction[String, SensorReading, String] {
  var myState: ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("myState", classOf[Int]))
  }

  override def processElement(value: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    context.getCurrentKey
    context.timestamp
    val service: TimerService = context.timerService
    service.currentWatermark
    service.currentProcessingTime
    service.registerEventTimeTimer(context.timestamp + 60000)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    // 处理定时器逻辑
  }
}







